package top.jacktgq.concurrent_utils.collection;
import java.io.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.BiConsumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @Author CandyWall
 * @Date 2020/11/6--12:11
 * @Description 多线程下使用线程安全的集合统计单词出现的个数
 */
public class WordCountDemo {
    public static void main(String[] args) {
        //generateTestData();
        wordCount4();
    }

    //版本1：
    //使用普通的HashMap集合，会有线程安全问题
    private static void wordCount1() {
        wordCount(
                () -> new HashMap<String, Integer>(),
                (map, words) -> {
                    synchronized (map) {
                        for (String word : words) {
                            //检查key有没有
                            Integer counter = map.get(word);
                            int newValue = counter == null ? 1 : counter + 1;
                            //没有则put
                            map.put(word, newValue);
                        }
                    }
                }
        );
    }

    //版本2：
    //使用线程安全的ConcurrentHashMap来替换HashMap，但是结果依然不正确，
    //原因：线程安全的集合类中的基础方法get()和put()单独来看是线程安全的，
    //但是将它们组合在一起用就不能保证线程安全了。
    private static void wordCount2() {
        wordCount(
                () -> new ConcurrentHashMap<String, Integer>(),
                (map, words) -> {
                    for (String word : words) {
                        //检查key有没有
                        Integer counter = map.get(word);
                        int newValue = counter == null ? 1 : counter + 1;
                        //没有则put
                        map.put(word, newValue);
                    }
                }
        );
    }

    //版本3：
    //将单词统计中对map的put()和get()方法整体添加synchronized关键字进行保护
    //能够得到正确的结果，但是性能却不高，ConcurrentHashMap内部用的就是细粒度的锁，
    //给它又加上了synchronized这个重量级锁之后，性能有降低了，这种做法不推荐！！！
    private static void wordCount3() {
        wordCount(
                () -> new ConcurrentHashMap<String, Integer>(),
                (map, words) -> {
                    synchronized (map) {
                        for (String word : words) {
                            //检查key有没有
                            Integer counter = map.get(word);
                            int newValue = counter == null ? 1 : counter + 1;
                            //没有则put
                            map.put(word, newValue);
                        }
                    }
                }
        );
    }

    //版本4：
    //使用ConcurrentHashMap内部的
    private static void wordCount4() {
        wordCount(
                () -> new ConcurrentHashMap<String, LongAdder>(),
                (map, words) -> {
                    for (String word : words) {
                        /*//检查key有没有
                        Integer counter = map.get(word);
                        int newValue = counter == null ? 1 : counter + 1;
                        //没有则put
                        map.put(word, newValue);*/
                        //如果缺少一个key，则计算生成一个value，然后将key value放入map
                        //也就是查询map中有没有这个word，如果有就直接把这个值返回，如果没有就新建一个LongAdder
                        LongAdder value = map.computeIfAbsent(word, (key) -> new LongAdder());
                        //执行累加
                        value.increment();

                    }
                }
        );
    }

    private static <V> void wordCount(Supplier<Map<String, V>> supplier, BiConsumer<Map<String, V>, List<String>> consumer) {
        //key   value
        //a     200
        //b     100
        Map<String, V> counterMap = supplier.get();
        ArrayList<Thread> threadList = new ArrayList<>();
        for (int i = 0; i < 26; i++) {
            int index = i + 1;
            threadList.add(new Thread(() -> {
                List<String> words = readFromFile(index);
                consumer.accept(counterMap, words);
            }));
        }
        threadList.forEach(t -> t.start());
        threadList.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(counterMap);
    }

    //从文件中获取所有的单词
    private static List<String> readFromFile(int index) {
        ArrayList<String> words = new ArrayList<String>();
        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new FileReader("tmp/" + index + ".txt"));
            String line = null;
            while ((line = reader.readLine()) != null) {
                words.add(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                reader.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return words;
    }


    //生成测试数据
    //生成方法：
    //将26个字母每个字母重复200次，总共5200个字母，然后将它们打乱顺序，再分别存放到26个文件中保存，
    //每一行存一个单词（为了简单，一个单词仅一个字母）
    private static void generateTestData() {
        String ALPHA = "abcdefghijklmnopqrstuvwxyz";
        int length = ALPHA.length();
        int count = 200;    //每个字母出现的次数
        ArrayList<String> list = new ArrayList<>(length * count);
        //依次遍历26个字母
        for (int i = 0; i < length; i++) {
            //取出每个字母
            char ch = ALPHA.charAt(i);
            //将每个字母重复count次存入list中
            for (int j = 0; j < count; j++) {
                list.add(String.valueOf(ch));
            }
        }
        //将list集合打乱
        Collections.shuffle(list);
        //将list集合分成26份，分别存到26个文件中，每个文件200个字母，一行一个字母
        for (int i = 0; i < 26; i++) {
            try (PrintWriter out =
                     new PrintWriter(new OutputStreamWriter(
                             new FileOutputStream("tmp/" + (i + 1) + ".txt")))) {
                String collect = list.subList(i * count, (i + 1) * count).stream()
                        .collect(Collectors.joining("\n"));
                out.print(collect);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
